package beautiful.butterfly.drds.data_exchange;


import beautiful.butterfly.drds.data_exchange.constant.Constants;
import beautiful.butterfly.drds.data_exchange.data_exchange.record.column.column_cast.ColumnCast;
import beautiful.butterfly.drds.data_exchange.job.JobContainer;
import beautiful.butterfly.drds.data_exchange.statistics.PerfTrace;
import beautiful.butterfly.drds.data_exchange.taskgroup.TaskGroupContainer;
import beautiful.butterfly.drds.data_exchange.util.ConfigParser;
import beautiful.butterfly.drds.data_exchange.util.Configuration;
import beautiful.butterfly.drds.data_exchange.util.ConfigurationValidate;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;


@Slf4j
public class DataExchangeService
{


    // 注意屏蔽敏感信息
    public static String filterJobConfiguration(final Configuration configuration)
    {
        Configuration jobConfWithSetting = configuration.getConfigureDataMap("job").clone();

        Configuration jobContent = jobConfWithSetting.getConfigureDataMap("content");

        filterSensitiveConfiguration(jobContent);

        jobConfWithSetting.set("content", jobContent);

        return jobConfWithSetting.beautify();
    }

    public static Configuration filterSensitiveConfiguration(Configuration configuration)
    {
        Set<String> keys = configuration.getKeys();
        for (final String key : keys)
        {
            boolean isSensitive = StringUtils.endsWithIgnoreCase(key, "password")
                    || StringUtils.endsWithIgnoreCase(key, "accessKey");
            if (isSensitive && configuration.get(key) instanceof String)
            {
                configuration.set(key, configuration.getString(key).replaceAll(".", "*"));
            }
        }
        return configuration;
    }

    public static void main(String[] args) throws Exception
    {
        dataExchange("xx");

    }

    /**
     * 首先需要解决一个问题,该jar被多个服务器共同使用,一个jvm或者说model中不能共享
     */
    public static void dataExchange(@NonNull String string)
    {
        log.info("输入参数:" + string);

        String jobPath = null;
        // 如果用户没有明确指定jobid, 则 datax.py 会指定 jobid 默认值为-1
        String jobIdString = null;


        Configuration configuration = ConfigParser.parse(jobPath);

        long jobId;
        if (!"-1".equalsIgnoreCase(jobIdString))
        {
            jobId = Long.parseLong(jobIdString);
        } else
        {
            // only for dsc & ds & datax 3 update
            String dscJobUrlPatternString = "/instance/(\\d{1,})/config.xml";
            String dsJobUrlPatternString = "/inner/job/(\\d{1,})/config";
            String dsTaskGroupUrlPatternString = "/inner/job/(\\d{1,})/taskGroup/";
            List<String> patternStringList = Arrays.asList(dscJobUrlPatternString,
                    dsJobUrlPatternString, dsTaskGroupUrlPatternString);
            jobId = parseJobIdFromUrl(patternStringList, jobPath);
        }


        configuration.set(Constants.container_job_id, jobId);


        log.info("\n" + DataExchangeService.filterJobConfiguration(configuration) + "\n");

        log.debug(configuration.toJSON());

        ConfigurationValidate.doValidate(configuration);
        DataExchangeService dataExchangeService = new DataExchangeService();
        dataExchangeService.start(configuration);
    }

    /**
     * -1 表示未能解析到 jobId
     * <p>
     * only for dsc & ds & datax 3 update
     */
    private static long parseJobIdFromUrl(List<String> patternStringList, String url)
    {
        long result = -1;
        for (String patternString : patternStringList)
        {
            result = doParseJobIdFromUrl(patternString, url);
            if (result != -1)
            {
                return result;
            }
        }
        return result;
    }

    private static long doParseJobIdFromUrl(String patternString, String url)
    {
        Pattern pattern = Pattern.compile(patternString);
        Matcher matcher = pattern.matcher(url);
        if (matcher.find())
        {
            return Long.parseLong(matcher.group(1));
        }

        return -1;
    }


    /* check job model (job/task) first */
    public void start(Configuration configuration)
    {

        // 绑定column转换信息
        ColumnCast.bind(configuration);

        /**
         * 初始化PluginLoader，可以获取各种插件配置
         */
        //Loaders.bind(allConf);

        boolean isJob = !("taskGroup".equalsIgnoreCase(configuration
                .getString(Constants.container_model)));
        //JobContainer会在schedule后再行进行设置和调整值
        int channelNumber = 0;
        Container container;
        long instanceId;
        int taskGroupId = -1;
        if (isJob)
        {

            container = new JobContainer(configuration);
            instanceId = configuration.getLong(
                    Constants.container_job_id, 0);

        } else
        {
            container = new TaskGroupContainer(configuration);
            instanceId = configuration.getLong(
                    Constants.container_job_id);
            taskGroupId = configuration.getInt(
                    Constants.container_taskgroup_id);
            channelNumber = configuration.getInt(
                    Constants.container_taskgroup_channel);
        }

        //缺省打开perfTrace
        boolean traceEnable = configuration.getBoolean(Constants.container_trace_enable, true);
        boolean perfReportEnable = configuration.getBoolean(Constants.report_datax_perflog, true);

        //standlone模式的datax shell任务不进行汇报
        if (instanceId == -1)
        {
            perfReportEnable = false;
        }

        int priority = 0;
        try
        {
            priority = Integer.parseInt(System.getenv("SKYNET_PRIORITY"));
        } catch (NumberFormatException e)
        {
            log.warn("prioriy set to 0, because NumberFormatException, the value is: " + System.getProperty("PROIORY"));
        }

        Configuration jobInfoConfig = configuration.getConfigureDataMap(Constants.job_jobinfo);
        //初始化PerfTrace
        PerfTrace perfTrace = PerfTrace.getInstance(isJob, instanceId, taskGroupId, priority, traceEnable);
        perfTrace.setJobInfo(jobInfoConfig, perfReportEnable, channelNumber);
        container.run();

    }

}
